package net.bwie.realtime.jtp.ads.log.job;

import net.bwie.realtime.jtp.ads.log.utils.AnalyzerUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import java.util.List;

public class EventPayCountJob {
    public static void main(String[] args) {
        //1.表执行环境
        TableEnvironment tabEnv = getTableEnv();
        //2.输入表-input：映射到kafka消息队列
        createInputTable(tabEnv);
        //3.数据处理-select
//        Table reportTable = handle(tabEnv);
        //4.输出表-output：映射到doris表
//        createOutputTable(tabEnv);
        //5.保存数据-insert
//        saveToSink(tabEnv,reportTable);
    }

    private static void saveToSink(TableEnvironment tabEnv, Table reportTable) {

    }

    private static void createOutputTable(TableEnvironment tabEnv) {

    }

    private static Table handle(TableEnvironment tabEnv) {
        return null;
    }

    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE dws_event_pay_transaction(\n" +
                "    `endTime` VARCHAR(50) COMMENT '结束时间',\n" +
                "    `liveRoomId` VARCHAR(50) COMMENT '直播间id',\n" +
                "    `orderCount` VARCHAR(50) COMMENT '订单数',\n" +
                "    `startTime` VARCHAR(50) COMMENT '开始时间',\n" +
                "    `userCount` VARCHAR(50) COMMENT '订单人数'\n" +
                ")WITH (\n" +
                "     'connector' = 'kafka',\n" +
                "     'topic' = 'dws_transaction_count',\n" +
                "     'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "     'properties.group.id' = 'gid_dws_traffic_search_keyword',\n" +
                "     'scan.startup.mode' = 'earliest-offset',\n" +
                "     'format' = 'json',\n" +
                "     'json.ignore-parse-errors' = 'true',\n" +
                "     'json.fail-on-missing-field' = 'false'\n" +
                "     )");
    }

    private static TableEnvironment getTableEnv() {
        //1.环境属性配置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        //2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone","Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism","1");
        configuration.setString("table.exec.state.ttl","5 s");
        configuration.setString("execution.checkpointing.interval","30 s");
        return tabEnv;
    }

    @FunctionHint(output = @DataTypeHint("ROW<keyword STRING>"))
    public static class IKAnalyzerFunction extends TableFunction<Row> {
        public void eval(String fullWord) throws Exception{
            //中文分词
            List<String> list = AnalyzerUtil.ikAnalyzer(fullWord);
            //循环遍历输出
            for (String keyword : list) {
                collect(Row.of(keyword));
            }
        }
    }
}


